package com.zhss.dfs.namenode.server;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 这个组件，就是负责管理集群里的所有的datanode的
 * @author zhonghuashishan
 *
 */
public class DataNodeManager {

	public DataNodeManager(){
		// 启动就开始监控问题线程
		new DataNodeHealthMonitor().start();
	}

	/**
	 * 内存中维护的datanode列表
	 */
	private Map<String,DataNodeInfo> dataNodes = new ConcurrentHashMap<String, DataNodeInfo>();
	
	/**
	 * datanode进行注册
	 * @param ip 
	 * @param hostname
	 */
	public Boolean register(String ip, String hostname) {
		DataNodeInfo datanode = new DataNodeInfo(ip, hostname);
		dataNodes.put(ip + "-" + hostname, datanode);
		return true;
	}


	/**
	 *  进行心跳检测
	 * @param ip
	 * @param hostname
	 * @return
	 */
	public Boolean heartBeat(String ip, String hostname) {
		DataNodeInfo dataNodeInfo = dataNodes.get(ip + "-" + hostname);
		dataNodeInfo.setLatestHeatBeatTime(System.currentTimeMillis());
		return true;
	}

	/**
	 *  监控dataNode是否健康
	 */
	class DataNodeHealthMonitor extends Thread{

		@Override
		public void run() {
			try{
				while (true) {
					ArrayList<String> unHealthDataNodes = new ArrayList<String>();
					for (DataNodeInfo dataNodeInfo : dataNodes.values()) {
						if (System.currentTimeMillis() - dataNodeInfo.getLatestHeatBeatTime() > 90 * 1000) {
							//当前dataNode存在问题
							unHealthDataNodes.add(dataNodeInfo.getIp() + "-" + dataNodeInfo.getHostname());
						}
					}
					if (!unHealthDataNodes.isEmpty()) {
						// 不健康节点进行移除
						for (String unHealthDataNode : unHealthDataNodes) {
							dataNodes.remove(unHealthDataNode);
						}
					}
					// 每隔30秒检测一次
					Thread.sleep(30 * 1000);
				}
			} catch (Exception e){
				e.printStackTrace();
			}
		}
	}
}
